package com.audaque.springboot.foshanupload.reactor.webfluxdemo.controller;


import com.alibaba.ttl.TransmittableThreadLocal;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.result.view.Rendering;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * @author zgb
 * @desc ...
 * @date 2022-11-24 17:45:26
 */

@Slf4j
@RestController
@RequestMapping("reactor")
public class ReactorController {

    @GetMapping("reactor2atomicReference")
    public String reactor2atomicReference() {
        Mono<String> hello = Mono.just("Hello");
        AtomicReference<String> atomicReference = new AtomicReference<String>();
        hello.subscribe(e -> atomicReference.set(e));
        String s = atomicReference.get();
        log.debug("s:" + s);
        AtomicReference<String> atomicReference2 = new AtomicReference<String>();
        hello.subscribe(e -> atomicReference2.set(e));
        String s2 = atomicReference2.get();
        log.debug("s2:" + s2);
        return s2;
    }

    @GetMapping("transmittableThreadLocal")
    public Mono<Void> transmittableThreadLocal() {
        TransmittableThreadLocal<String> transmittableThreadLocal = new TransmittableThreadLocal<>();
        transmittableThreadLocal.set("1");
        String s = transmittableThreadLocal.get();
        log.debug("s:" + s);

        Mono<Integer> mono = Mono.just(1);
        //使用subscribe方法订阅  :Disposable subscribe(Consumer<? super T> consumer)
        mono.subscribe(e -> {
            System.out.println(e);
            String s1 = transmittableThreadLocal.get();
            log.debug("s1:" + s1);
        });

        Flux<Integer> just = Flux.just(1, 2, 3, 4);
        just.subscribe(e -> {
            System.out.println(e);
            String s2 = transmittableThreadLocal.get();
            log.debug("s2:" + s2);
        });


        return Mono.empty();
    }

    @GetMapping("mapSubscribe")
    public Mono<Void> mapSubscribe() {
        Mono.just("Hello")
                .map(s -> "1")
                .subscribe(e -> log.debug("e:" + e));
        return Mono.empty();
    }

    @GetMapping("rendering")
    public Mono<Rendering> rendering() {
        return Mono.just(Rendering.redirectTo( "www.baidu.com").build());
    }


    @GetMapping("empty")
    public Mono<Void> empty() {
        return Mono.empty();
    }

    @GetMapping("toReactor")
    public void toReactor() {
        //创建生产者

        Flux<Integer> just = Flux.just(1, 2, 3, 4);
        Mono<Integer> mono = Mono.just(1);

        Mono.create(monoSink -> monoSink.success("hell0"));

        Integer[] array = {1, 2, 3, 4};
        Flux.fromArray(array);

        List<Integer> list = Arrays.asList(array);
        Flux.fromIterable(list);

        Stream<Integer> stream = list.stream();
        Flux.fromStream(stream);
    }

    @GetMapping("subscribe")
    public void subscribe() {
        Mono<Integer> mono = Mono.just(1);
        //使用subscribe方法订阅  :Disposable subscribe(Consumer<? super T> consumer)
        mono.subscribe(e -> System.out.println(e));
    }

    @GetMapping("print")
    public void print() {
        Mono<Integer> mono = Mono.just(1);
        //使用subscribe方法订阅
        mono.subscribe(System.out::print);
    }


    @GetMapping("baseSubscriber")
    public void baseSubscriber() {

        //创建生产者
        Flux flux = Flux.range(1, 10);
        //创建消费者
        Subscriber subscriber = new BaseSubscriber<Integer>() {
            protected void hookOnNext(Integer value) {
                System.out.println(Thread.currentThread().getName() + " -> " + value);
                //建立关系时设置接收数量大于1代表开始接收消息，设置小于1会报错
                request(1);
            }
        };
        //订阅关系
        flux.subscribe(subscriber);
        Flux flux1 = Flux.range(1, 10);
    }
    //Sink可以理解为数据池，负责存储数据，根据功能不同划分，如IgnoreSink，BufferAsyncSink，LatestAsyncSink。
    //Sink#next会将数据放入池中，由Sink缓存或直接发送给订阅者。

    @GetMapping("generate")
    public void generate() {

        Flux.generate(sink -> {
                    int k = (int) (Math.random() * 10);
                    if (k > 8)
                        sink.complete();
                    sink.next(k);
                })
                .subscribe(i -> {
                    System.out.println("receive:" + i);
                });

    }

    @GetMapping("create")
    public void create() {
        Flux.create(sink -> {
            System.out.println("please entry data");
            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
            while (true) {
                try {
                    sink.next(br.readLine());
                } catch (IOException e) {
                }
            }
        }).subscribe(i -> {
            System.out.println("receive:" + i);
        });
    }

    /**
     * Schedulers提供的静态方法可以创建不同的线程执行环境。
     * Schedulers.immediate() 直接在当前线程执行
     * Schedulers.single() 在一个重复利用的线程上执行
     * Schedulers.boundedElastic() 在由Reactor维护的线程池上执行，该线程池中闲置时间过长（默认值为60s）的线程也将被丢弃，创建线程数量上限默认为CPU核心数x 10。线程数达到上限后，最多可提交10万个任务，这些任务在线程可用时会被执行。该线程池可以为阻塞操作提供很好的支持。阻塞操作可以执行在独立的线程上，不会占用其他资源。
     * Schedulers.parallel() 固定线程，对于异步IO，可以使用该方案。
     *
     * @throws InterruptedException
     */
    @GetMapping("parallel")
    public void parallel() throws InterruptedException {
        Flux.range(0, 100)
                .parallel()
                .runOn(Schedulers.parallel())
                .subscribe(i -> {
                    System.out.println(Thread.currentThread().getName() + " -> " + i);
                });

    }

    @GetMapping("publishOn")
    public void publishOn() throws InterruptedException {
        Flux.range(1, 3)
                .doOnNext(i -> {
                    System.out.println(Thread.currentThread().getName() + " doOnNext:" + i);
                })
                //publishOn影响当前操作符方法后面操作的线程执行环境
                .publishOn(Schedulers.newParallel("myParallel"))
                .skip(1)
                .subscribe(i -> {
                    System.out.println(Thread.currentThread().getName() + " receive:" + i);
                });
    }

    @GetMapping("subscribeOn")
    public void subscribeOn() throws InterruptedException {
        Flux.range(1, 3)
                .doOnNext(i -> {
                    System.out.println(Thread.currentThread().getName() + " doOnNext:" + i);
                })
                .subscribeOn(Schedulers.newParallel("myParallel"))
                .skip(1)
                //subscribeOn则影响整个链路的线程执行环境。
                .subscribe(i -> {
                    System.out.println(Thread.currentThread().getName() + " receive:" + i);
                });
    }


    @GetMapping("backpressure")
    public void backpressure() throws InterruptedException {
        Flux.<Integer>create(sink -> {
                    for (int i = 0; i < 50; i++) {
                        System.out.println("push: " + i);
                        sink.next(i);
                        try {
                            Thread.sleep(10);
                        } catch (InterruptedException e) {
                        }
                    }
                    /**例如我的服务器可以同时处理 2000 个用户请求，那么我就把请求上限设置为 2000，这个 2000 就是我的 Buffer，当超出 2000 的时候，就产生了 Backpressure。
                     * FluxSink.OverflowStrategy定义了在背压场景下的几种处理策略。
                     * IGNORE 完全忽略新的数据
                     * ERROR Publisher抛出异常,Sink关闭，后面的数据被抛弃,不会被消费
                     * DROP 抛弃数据，触发Flux#onBackpressureDrop方法
                     * LATEST 订阅者只能获取最新的一个数据
                     * BUFFER 缓存所有的数据，注意，该缓存没有边界，可能导致内存溢出
                     * FluxSink.OverflowStrategy类似于线程池的任务拒绝策略。
                     * @throws InterruptedException
                     */
                }, FluxSink.OverflowStrategy.ERROR)
                //publishOn方法指定后续运行线程环境
                .publishOn(Schedulers.newSingle("receiver"), 1)
                .subscribe(new BaseSubscriber<Integer>() {
                    protected void hookOnSubscribe(Subscription subscription) {
                        subscription.request(1);
                    }

                    protected void hookOnNext(Integer value) {
                        System.out.println("receive:" + value);
                        try {
                            Thread.sleep(12);
                        } catch (InterruptedException e) {
                        }
                        request(1);
                    }

                    protected void hookOnError(Throwable throwable) {
                        throwable.printStackTrace();
                        System.exit(1);
                    }
                });
        new CountDownLatch(1).await();
    }


    @GetMapping("mono2")
    public Mono<String> mono2() {
        //异步
        //存放0-1个数据流序列，用于响应单个对象
        Mono<String> from = Mono.fromSupplier(() -> {
            try {
                //模拟处理时间为5秒
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "mono2:在5秒中之前已经处理完整个方法，5秒钟后返回数据，证明是异步调用：" + LocalDateTime.now();
        });
        //上面是异步，所以下面不用等，马上打印
        System.out.println("mono2:方法结束:" + "thread : " + Thread.currentThread().getName() + " ===  " + LocalDateTime.now());
        //处理完业务逻辑后通过回调方式响应给浏览器。
        return from;
    }

    /**
     * 默认：Accept:text/html,application/xhtml+xml,application/xml;q=0.8,application/signed-exchange;v=b3;q=0.7
     * 服务器端事件流的语法是非常简单的。你需要把 "Content-Type" 报头设置为 "text/event-stream"。现在，您可以开始发送事件流了。
     *
     * @return
     */
    @GetMapping(value = "flux2", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> flux2() {
        //存放0-n个数据流序列,用于响应集合
        Flux<String> stringFlux = Flux.fromStream(IntStream.range(1, 6).mapToObj(i -> {
            try {
                //模拟处理时间为1秒
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "flux2：" + i + "date time: " + LocalDateTime.now();
        }));
        System.out.println("flux2：方法结束:thread : " + Thread.currentThread().getName() + " ===  " + LocalDateTime.now());
        return stringFlux;

    }

    @GetMapping(value = "mono3", produces = MediaType.APPLICATION_JSON_VALUE)
    public String mono3() {
        //不支持响应式阻塞
        return mono2().block();

    }


    @GetMapping(value = "flux3", produces = MediaType.APPLICATION_JSON_VALUE)
    public List<String> flux3() {
        //不支持响应式阻塞
        return flux2().toStream().collect(Collectors.toList());
    }


}
